package com.example.kafkademo.kafkaSpark;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Iterator;

public class JavaClusterWordCount {

    public static void main(String[] args) {
/**
 * 第一步，创建Spark的配置对象SparkConf，设置Spark程序的运行时的配置信息，
 * 例如说通过setMaster来设置程序要链接的Spark集群的Master的URL，
 * 如果设置为local，则代表Spark程序在本地运行，特别适合于机器配置较差的情况
 */
        SparkConf sparkConf = new SparkConf().setAppName("LocalWordCountByJava").setMaster("local");
/**
 * 第二步，创建SparkContext对象
 * SparkContext是Spark程序所有功能的唯一入口，无论是采用Scala，java，python，R等都
 * 必须有一个SparkContext（不同语言具体类名称不同，如果是Java的话，则为JavaSparkContext）
 * 同时还会负责Spark程序在Master注册程序等
 * SparkContext是整个Spark应用程序至关重要的一个对象
 */
        JavaSparkContext jsc = new JavaSparkContext(sparkConf);//其底层实际上是Scala的SparkContext
/**
 * 第三步，根据具体的数据来源（HDFS，HBase，Local，FS，DB，S3等），通过JavaSparkContext来创建JavaRDD
 * JavaRDD的创建方式有三种：根据外部数据来源（例如HDFS），
 * 根据Scala集合，由其他的RDD操作数据会将RDD划分成一系列Partition，
 * 分配到每个Partition的数据属于一个Task处理范畴
 */
        JavaRDD<String> lines = jsc.textFile("hdfs://192.168.168.200:9000/input/words.txt");
//如果是Scala，由于SAM转化，所以可以写成val words=lines.flatMap{line =>line.split(" ")}
        JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterator<String> call(String line) {
                return Arrays.asList(line.split(" ")).iterator();
            }
        });

/**
 * 第4步：对初始的JavaRDD进行Transformation级别的处理，例如map，filter等高阶函数等的编程，来进行具体的数据计算
 * 第4.1步：在单词拆分的基础上对每个单词实例进行计数为1，也就是word =>(word,1)
 */
        JavaPairRDD<String, Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() {
            public Tuple2<String, Integer> call(String word) throws Exception{
                return new Tuple2<String, Integer>(word, 1);
            }
        });
/**
 * 统计总次数
 */
        JavaPairRDD<String,Integer> wordCount=pairs.reduceByKey(new Function2<Integer,Integer,Integer>()
        {
            public Integer call(Integer v1,Integer v2)throws Exception
            {
                return v1+v2;

            }
        });

        wordCount.foreach(new VoidFunction<Tuple2<String,Integer>>(){
            public void call(Tuple2<String,Integer> pairs) throws Exception {
                System.out.println(pairs._1()+":"+pairs._2());
            }
        });

        jsc.close();
    }
}
